package com.sojson.util.mq.rabbit.immit.test;

import java.io.IOException;

import javax.annotation.PostConstruct;

import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitOperations;
import org.springframework.amqp.rabbit.core.RabbitOperations.OperationsCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Payload;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.sojson.constant.ConstantCache;
import com.sojson.util.StringUtil;
import com.sojson.util.mq.rabbit.code.test.RabbitCodeTestUtil;
import com.sojson.util.mq.rabbit.immit.RabbitImmitUtil;
import com.sojson.util.mq.rabbit.immit.config.RabbitConfig;
import com.sojson.util.mq.rabbit.immit.interfaces.impl.RabbitDelayMsgUtilImpl.RabbitDirectDelayMsgUtilImpl;

import lombok.extern.slf4j.Slf4j;

/**
 * Rabbit消息队列测试类</BR>
 * UI地址http://192.168.100.101:15672</BR>
 * UI默认账号root</BR>
 * UI默认密码123456</BR>
 * 
 * @author liu
 * @date 2021-07-11
 */
@Slf4j
public class RabbitImmitTestUtil {

    @Autowired
    private AmqpAdmin amqpAdmin;
    @Autowired
    private RabbitTemplate template;

    /**
     * 声明交换机和队列
     * 
     * @return
     * @throws IOException
     */
    @PostConstruct
    public void init() throws IOException {
        // 获取要声明的类型的工具类
        RabbitDirectDelayMsgUtilImpl instances = RabbitImmitUtil.getDelay().getDirect();

        // 声明交换机
        instances.exchangeDeclare(amqpAdmin, RabbitCodeTestUtil.EXCHANGE_NAME1);

        // 声明队列
        instances.queueDeclare(amqpAdmin, RabbitCodeTestUtil.QUEUE_NAME1);
        // 将交换机和队列绑定
        instances.queueBind(amqpAdmin, RabbitCodeTestUtil.EXCHANGE_NAME1, RabbitCodeTestUtil.QUEUE_NAME1);

        // 声明队列
        instances.queueDeclare(amqpAdmin, RabbitCodeTestUtil.QUEUE_NAME2);
        // 将交换机和队列绑定
        instances.queueBind(amqpAdmin, RabbitCodeTestUtil.EXCHANGE_NAME1, RabbitCodeTestUtil.QUEUE_NAME2);
    }

    /** 生产者(无须确认) */
    public void sc1() throws Exception {
        RabbitDirectDelayMsgUtilImpl instances = RabbitImmitUtil.getDelay().getDirect();
        instances.basicPublish(template, RabbitCodeTestUtil.EXCHANGE_NAME1, StringUtil.EMPTY, "这里是消息");
    }

    /** 生产者(发布确认) */
    public void sc2() throws Exception {
        RabbitDirectDelayMsgUtilImpl instances = RabbitImmitUtil.getDelay().getDirect();

        Boolean invoke = template.invoke(new OperationsCallback<Boolean>() {
            @SuppressWarnings("synthetic-access")
            @Override
            public Boolean doInRabbit(RabbitOperations operations) {
                try {
                    instances.basicPublish(operations, RabbitCodeTestUtil.EXCHANGE_NAME1, StringUtil.EMPTY, "这里是消息");
                } catch (IOException e) {
                    log.error(e.getMessage(), e);
                }
                return operations.waitForConfirms(3000);
            }
        });
        System.out.println(invoke);

        invoke = template.invoke(new OperationsCallback<Boolean>() {

            @Override
            @SuppressWarnings("synthetic-access")
            public Boolean doInRabbit(RabbitOperations operations) {
                try {
                    instances.basicPublish(operations, RabbitCodeTestUtil.EXCHANGE_NAME1, StringUtil.EMPTY, "这里是消息");
                } catch (IOException e) {
                    log.error(e.getMessage(), e);
                }
                return operations.waitForConfirms(3000);
            }
        },
            // 同步确认回调,需在doInRabbit方法中暂停到服务器返回信息才会调调用
            new ConfirmCallback() {
                @Override
                public void handle(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("确认回调: ");
                    System.out.println("DeliveryTag:-标签: " + deliveryTag);
                }
            },
            // 同步失败回调,需在doInRabbit方法中暂停到服务器返回信息才会调调用
            new ConfirmCallback() {
                @Override
                public void handle(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("失败回调: ");
                    System.out.println("DeliveryTag:-标签: " + deliveryTag);
                }
            });
        System.out.println(invoke);
    }

    /**
     * 消费者(自动确认)</BR>
     * 必须指定containerFactory,不然我们自定义的配置会失效</BR>
     */
    @RabbitHandler
    @RabbitListener(queues = ConstantCache.KEY_P_PREFIX + RabbitCodeTestUtil.QUEUE_NAME1,
        ackMode = RabbitImmitUtil.ACK_MODE_AUTO, containerFactory = RabbitConfig.LISTENER_KEY)
    public void xf11(@Payload String body, Channel channel, Message message) throws IOException {
        System.out.println("1111111111");
    }

    /**
     * 消费者(手动确认)</BR>
     * 必须指定containerFactory,不然我们自定义的配置会失效</BR>
     */
    @RabbitHandler
    @RabbitListener(queues = ConstantCache.KEY_P_PREFIX + RabbitCodeTestUtil.QUEUE_NAME1,
        ackMode = RabbitImmitUtil.ACK_MODE_MANUAL, containerFactory = RabbitConfig.LISTENER_KEY)
    public void xf12(@Payload String body, Channel channel, Message message) throws IOException {
        System.out.println("2222222222");

        MessageProperties messageProperties = message.getMessageProperties();

        try {
            // 在处理完消息后手动进行确认,如果不确认,默认为返回失败
            /*
             * 参数1： 消费消息的index
             * 参数2： 是否批量确认消息,前提是在同一个channel里面,且是在该消息确认前没有被确认的消息才能批量确认
             */
            channel.basicAck(messageProperties.getDeliveryTag(), false);
        } catch (Exception e) {
            /**
             * deliveryTag: 消费消息的index
             * requeue: 对异常消息的处理,true表示重排序,false表示丢弃
             */
            channel.basicReject(messageProperties.getDeliveryTag(), false);
            /**
             * deliveryTag: 消费消息的index
             * multiple: 是否批量否定多个消息,设为false就与basicReject功能一样,true的前提也是在同一个channel,且在该消息否定前存在未确认的消息
             * requeue: 对异常消息的处理,true表示重排序,false表示丢弃
             */
            // channel.basicNack(messageProperties.getDeliveryTag(), false, true);
        }
    }

    /**
     * 消费者(自动确认)</BR>
     * 必须指定containerFactory,不然我们自定义的配置会失效</BR>
     */
    @RabbitHandler
    @RabbitListener(queues = ConstantCache.KEY_P_PREFIX + RabbitCodeTestUtil.QUEUE_NAME2,
        ackMode = RabbitImmitUtil.ACK_MODE_AUTO, containerFactory = RabbitConfig.LISTENER_KEY)
    public void xf2(@Payload String body, Channel channel, Message message) throws IOException {
        System.out.println("3333333333");
    }

}